package com.itliang.mqtt.service;

import com.itliang.mqtt.model.property.MqttProperty;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.List;

@Service
@Slf4j
public class MqttService {

    @Resource
    private MqttProperty mqttProperty;

    //mqtt连接
    private MqttClient mqttClient;

    /**
     * 配置MQTT连接信息
     *
     * @return
     */
    public MqttConnectOptions getOptions() {
        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
        // 是否清除Session
        mqttConnectOptions.setCleanSession(mqttProperty.getCleanSession());
        // MQTT服务器用户名
        mqttConnectOptions.setUserName(mqttProperty.getUsername());
        // MQTT服务器密码
        mqttConnectOptions.setPassword(mqttProperty.getPassword().toCharArray());
        // MQTT连接超时时间
        mqttConnectOptions.setConnectionTimeout(mqttProperty.getConnectionTimeout());
        // 设置MQTT心跳时间
        mqttConnectOptions.setKeepAliveInterval(mqttProperty.getKeepAliveInterval());
        // 设置MQTT自动重连
        mqttConnectOptions.setAutomaticReconnect(mqttProperty.getAutomaticReconnect());
        return mqttConnectOptions;
    }

    /**
     * 连接MQTT服务端
     */
    public void connect() {
        //防止重复创建MQTTClient实例
        if (mqttClient == null) {
            try {
                mqttClient = new MqttClient(mqttProperty.getHost(), mqttProperty.getClientId(), new MemoryPersistence());
                //MqttDataService是接收订阅topic数据推送的回调
                mqttClient.setCallback(new MqttCallbackService());
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }
        MqttConnectOptions options = getOptions();
        //判断拦截状态
        if (!mqttClient.isConnected()) {
            try {
                mqttClient.connect(options);
                log.info("[MQTT] 连接成功！");
            } catch (MqttException e) {
                e.printStackTrace();
            }
        } else {
            try {
                mqttClient.disconnect();
                mqttClient.connect(options);
                log.info("[MQTT] 连接成功！");
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 启动程序并订阅Topic
     *
     * @PostConstruct 注解的作用是服务初始化启动完成后会主动执行有该注解的方法
     */
    @PostConstruct
    public void start() {
        log.info("MQTT连接数据初始化成功，开始连接....");
        try {
            connect();
            List<String> subTopics = mqttProperty.getSubTopics();
            List<Integer> qos = mqttProperty.getQos();
            for (int i = 0; i < subTopics.size(); i++) {
                mqttClient.subscribe(subTopics.get(i), qos.get(i));
            }
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 发送数据
     */
    public void publish(String topic, String content, int qos) throws MqttException {
        // MqttMessage message = new MqttMessage(content.getBytes());
        byte[] bytes = content.getBytes();
        try {
            mqttClient.publish(topic, bytes, qos, false);
        } catch (Exception e) {
            log.error("[MQTT]发送数据失败，原因 {}", e.getMessage());
        }
    }

    /**
     * 断线重连
     */
    public void reConnect() throws MqttException {
        if (null != mqttClient) {
            MqttConnectOptions options = getOptions();
            mqttClient.connect(options);
            log.info("[MQTT] 重连成功！");
        }
    }

}

